package cn.itcast.batch;


import cn.itcast.batch.utils.ConfigLoader;
import cn.itcast.batch.utils.DateUtil;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Types;
import java.util.Objects;
import java.util.UUID;

//多维度分析数据指标计算的公共类
//天，周，月的维度进行多维度分析数据准确率
public class AnalysisDataRate extends JDBCFormatAbstract{

    private Logger logger = LoggerFactory.getLogger(AnalysisDataRate.class);

    //指定按照天统计准确率
    private final static String DAY = DateUtil.getTodayDate();
    //指定按照周计算准确率
    private final static String WEEK = DateUtil.getNowWeekStart();
    //指定按照月计算准确率
    private final static String MONTH = DateUtil.getYearMonthDate()+"00";
    //指定统计条件
    private String condition;
    public AnalysisDataRate(String condition){
        this.condition = condition;
    }

    public void executeTask(){
        /**
         * 1.初始化flink批处理运行环境
         * 2.验证condition是否为空
         * 3.按照维度获取到数据，得到JDBCInputFormat对象
         * 4.将format对象添加到环境获取到数据
         * 5.将查询到的正常数据以及异常数据转换成写入数据的模型对象ROW
         * 6.获取到jdbcOutFormat对象
         * 7.将数据写入mysql数据库
         * 8.提交任务
         */
        try {

            //todo 1.初始化flink批处理运行环境
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
            //todo 2.验证condition是否为空
            Objects.requireNonNull(condition, "执行计算任务的时候必须要传递参数");
            //todo 3.按照维度获取到数据，得到JDBCInputFormat对象
            JDBCInputFormat hiveInputFormat = getHiveInputFormat();
            //todo 4.将format对象添加到环境获取到数据
            DataSource<Row> dataSource = env.createInput(hiveInputFormat);
            //todo 5.将查询到的正常数据以及异常数据转换成写入数据的模型对象ROW
            DataSet<Row> resultDataSet = convertHiveDataSource(dataSource);
            //todo 6.获取到jdbcOutFormat对象
            JDBCOutputFormat mysqlOutputFormat = getMysqlOutputFormat();
            //todo 7.将数据写入mysql数据库
            resultDataSet.output(mysqlOutputFormat);
            //todo 8.提交任务
            env.execute();
        }catch (Exception e){
            logger.error(e.getMessage());
            e.printStackTrace();
        }
    }

    //从hive获取到数据
    private JDBCInputFormat getHiveInputFormat(){
        //指定驱动名称
        String driverName = ConfigLoader.getProperty("hive.jdbc.driver");
        String url = ConfigLoader.getProperty("hive.jdbc.url");
        String userName = ConfigLoader.getProperty("hive.jdbc.user");
        String password = ConfigLoader.getProperty("hive.jdbc.password");
        String inputSql = "select srcTotalNum,errorTotalNum from " +
                "(select count(1) srcTotalNum from itcast_src where dt>='"+condition+"') src," +
                "(select count(1) errorTotalNum from itcast_error dt>='"+condition+"') error;";
        //指定查询sql语句返回的字段信息
        TypeInformation[] typeInformations = {BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO};
        String[] colNames = {"srcTotalNum", "errorTotalNum"};
        RowTypeInfo rowTypeInfo = new RowTypeInfo(typeInformations, colNames);
        return getBatchJDBCInputFormat(driverName,url,userName,password,inputSql,rowTypeInfo);
    }

    //获取到mysql的jdbc输出对象
    private JDBCOutputFormat getMysqlOutputFormat(){
        String driverName = ConfigLoader.getProperty("mysql.jdbc.driver");
        String url = ConfigLoader.getProperty("mysql.jdbc.url");
        String userName = ConfigLoader.getProperty("mysql.jdbc.user");
        String password = ConfigLoader.getProperty("mysql.jdbc.password");
        //定义表的集合数据
        String[] mysqlTable = new String[]{"itcast_data_rate_day","itcast_data_rate_week","itcast_data_rate_month"};

        String outputSql = null;

        switch (setDynamicValue(condition,mysqlTable)){
            case "itcast_data_rate_day":
                outputSql = "insert into itcast_data_rate_day(series_no, src_total_num, error_src_total_num, data_accuracy, data_error_rate,day,process_date)\n" +
                        " values(?,?,?,?,?,?,?);";
                break;
            case "itcast_data_rate_week":
                outputSql = "insert into itcast_data_rate_week(series_no, src_total_num, error_src_total_num, data_accuracy, data_error_rate,week,process_date)\n" +
                        " values(?,?,?,?,?,?,?);";
                break;
            case "itcast_data_rate_month":
                outputSql = "insert into itcast_data_rate_month(series_no, src_total_num, error_src_total_num, data_accuracy, data_error_rate,month,process_date)\n" +
                        " values(?,?,?,?,?,?,?);";
                break;

        }

        int[] sqlType = new int[]{Types.VARCHAR,Types.BIGINT,Types.BIGINT,Types.FLOAT,Types.FLOAT,Types.VARCHAR,Types.VARCHAR};
        return getBatchJDBCOutputFormat(driverName,url,userName,password,outputSql,sqlType);

    }


    /**
     * 根据查询条件返回对应的查询表的名称
     * @param condition
     * @param mysqlTables
     * @return
     */
    private String setDynamicValue(String condition,String[] mysqlTables){
        if(condition.equals(DAY)){
            return mysqlTables[0];
        }else if(condition.equals(WEEK)){
            return mysqlTables[1];
        }else if(condition.equals(MONTH)){
            return mysqlTables[2];
        }else {
            return mysqlTables[0];//默认是天
        }
    }


    /**
     * 根据正确数据量和错误数据量计算正确率和错误率
     * @param hiveDataSet
     * @return
     */
    public DataSet<Row> convertHiveDataSource(DataSet<Row> hiveDataSet){
        return hiveDataSet.map(row -> {
            //获取正确数据的数据量
            float srcTotalNum = Float.parseFloat(row.getField(0).toString());
            float errorTotalNum = Float.parseFloat(row.getField(1).toString());
            //正确率
            float dataAccuracy = srcTotalNum/(srcTotalNum+errorTotalNum);

            Row resultRow = new Row(7);
            resultRow.setField(0, UUID.randomUUID().toString());
            resultRow.setField(1,srcTotalNum);
            resultRow.setField(2,errorTotalNum);
            resultRow.setField(3,dataAccuracy); //正确率
            resultRow.setField(4,1-dataAccuracy);//错误率
            resultRow.setField(5,condition);
            resultRow.setField(6, DateUtil.getTodayDate()); //获取
            return resultRow;
        });
    }

}
